iT邦幫忙

2023 iThome 鐵人賽

DAY 29
0
Software Development

FastAPI 開發筆記:從新手到專家的成長之路系列 第 29

[Day 29] 定時任務與健康狀態檢查

  • 分享至 

  • xImage
  •  

最後想了想,決定來介紹一下怎麼寫「定時任務」和「健康狀態檢查 (health check)」這兩個小主題~

health check 翻成中文好不習慣...

定時任務

在某些時候,我們後端需要有一個定時執行的程式 (例如:向某個地方取得最新資料並更新到資料庫、清除過期 token),此時我們就需要寫一個定時任務來幫我們完成這件事。

定時任務的實作方法也不只一種,例如:

1. 使用 APScheduler

APScheduler是一個可以寫排程的的套件,我們可以在啟動 FastAPI 的時候同時也初始化 APScheduler,後續再讓他自己定時完成任務,也可以透過 API 來取得或修改 APScheduler 的排程。

這樣的做法比較像是用 FastAPI 做介接,讓我們可以透過打 API 的方式去隔空操作。

2. 使用 Celery 的 celery beat

概念跟上面那個差不多,就是用 beat_schedule 設定定時任務,詳細可以參考這個文件

3. 用 fastapi-utils 的 repeat_every

在開始介紹之前,先說明一下,這個好用模組已經有段時間沒更新了 (最後一次更新是 2023 年 3 月),因此與最新版本的 FastAPI 相依的模組有落差,例如:fastapi-utils 尚未支援 pydantic 2.0、SQLAlchemy 2.0,儘管有多個 issue (issue 1issue 2)希望可以支援新版,但作者似乎已經銷聲匿跡了...

但好在有其他熱心網友 fork 這個模組並更新支援的模組,例如:FastApi-RESTfulfastapi-utilities,因此大家在使用時可以考慮一下到底要安裝哪個套件。

這是我個人最喜歡的寫法,因為這個最簡潔,畢竟我只要定時啟動就好,不需要其他功能。

fastapi-utils 是一個專門提供 FastAPI 相關功能的模組 (相信從名稱就可以感受得出來),它裡面就有 repeat_every 這個 decorator 可以幫我們做到定時任務的設定。

首先,先安裝 fastapi-utils

pip install fastapi-utils

或是 FastApi-RESTful

pip install fastapi-restful[all]

接著看一下這個範例 (這邊用 FastApi-RESTful 當例子)

# main.py
from fastapi import FastAPI
# from fastapi_utils.tasks import repeat_every
from fastapi_restful.tasks import repeat_every

app = FastAPI()

@repeat_every(seconds=5)
async def repeat_task() -> None:
    print("test")

@app.on_event("startup")
async def startup_event():
    await repeat_task()

使用時,只要在執行的定時任務用 repeat_every 修飾即可,如果需要其他設定則可以參考文件。此外,通常我們也會希望啟動服務的時候也要執行一次,因此可以放進 startup 事件中 (或是用較新的 lifespan 寫法也可以)。

相信大家應該有感受到這個做法的簡潔吧 XD

健康狀態檢查

有時候,我們需要確認後端是否正常運作,這件事的時機可能是在部屬後 (快速確認是否部屬成功),也有可能是定時觀察 (避免服務掛了都不知道 XD)。最簡單的做法就是打一個最簡單的 API (或是故意寫一個 API 專門回傳極簡單的訊息),看有沒有如預期地拿到正確回應就好。

但是,如果我們的服務更複雜一些,還有接資料庫,那我們就需要再多確認後端是否可以連上資料庫。這件事也不難,就在 API 內加上一個 Depends(),讓 API 取得與資料庫連線的 Session (詳細可以參考 [Day 15])。

但是 (第二次 XD),如果服務再更複雜,有更多東西需要檢查,那這個 health check 用的 API 就會越來越複雜,這時,可以考慮使用 fastapi-health 這個套件。

這邊直接看官方範例

from fastapi import FastAPI, Depends
from fastapi_health import health

def get_session():
    return True

def is_database_online(session: bool = Depends(get_session)):
    return session

app = FastAPI()
app.add_api_route("/health", health([is_database_online]))

它的核心用法就是用 add_api_route() 的第二個變數這邊使用 health 這個函數,我們只要把要檢查的項目放進 list 傳給 health 函數就好了。其他進階設定就請大家再去看文件說明了 (一點點而已)。

補充

最後還是想補充一下,如果是檢查網站是否正常,或是後端的 API 有暴露在外的話,可以考慮使用 UptimeRobot 這個免費工具 (在 [Day 26] 有稍微提到過),他除了可以定時進行 health check 之外,也可以再發現問題的時候發送通知,通知的方式有很多種,從基本的 email 到常見的通訊軟體都可以 (透過 webhook),使用起來很方便~

小結

今天介紹了兩個小主題,在「定時任務」我們介紹了

  1. 多個實作的方法
  2. fastapi-utils 的範例

在「health check」則是簡單的介紹了 fastapi-health 這個套件並補充介紹 UptimeRobot 這個工具

終於只剩下最後一篇了,雖然還有很多主題可以寫,但明天還是用心得來收尾好了 XD
/images/emoticon/emoticon39.gif


上一篇
[Day 28] WebSocket 的實作與測試
下一篇
[Day 30] 文章回顧與完賽心得
系列文
FastAPI 開發筆記:從新手到專家的成長之路30
圖片
  直播研討會
圖片
{{ item.channelVendor }} {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言